package com.atguigu.stream.watermark


import java.sql.Timestamp

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object GenWatermark {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //
    val stream = env.socketTextStream("192.168.35.102", 9999, '\n')

    val str = stream
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      .assignTimestampsAndWatermarks (
        new MyAssigner
      )

      .keyBy(_._1)
      .timeWindow(Time.seconds(10))
      .process(new WindowResult)
    str.
      print()
    env.execute()
  }

class MyAssigner extends AssignerWithPeriodicWatermarks[(String,Long)]{
  //设置最大延迟时间
  val bound:Long = 10*1000L
  //系统观察到元素包含的最大时间戳
  var maxTs:Long = Long.MinValue+bound




  //产生水位线的逻辑
  //默认每隔200毫秒调用一次 - 最大延迟时间
  //可以设置调用时间
  override def getCurrentWatermark: Watermark = {
    new Watermark(maxTs - bound)
  }

  //定义抽取时间戳的逻辑，每到一个事件就调用一次
  override def extractTimestamp(t: (String, Long), l: Long): Long = {
    maxTs = maxTs.max(t._2)//   更新观察到的最大时间戳
    t._2    //  将抽取的时间戳返回

  }
}



  class WindowResult extends ProcessWindowFunction[(String,Long),String,String,TimeWindow]{
    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]): Unit = {
      out.collect(new Timestamp(context.window.getStart)+"-"+new Timestamp(context.window.getEnd)+"窗口中有"+elements.foreach(line => print(line))+"个元素")
    }
  }


}